package com.ikas.ai.framework.node.impl;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.ikas.ai.framework.core.event.NodeEvent;
import com.ikas.ai.framework.core.listener.NodeListener;
import com.ikas.ai.framework.core.listener.NodeLogListener;
import com.ikas.ai.framework.core.model.BatchDataFunction;
import com.ikas.ai.framework.core.model.NodeInputVariables;
import com.ikas.ai.framework.flow.ScoreFlow;
import com.ikas.ai.framework.node.Node;
import com.ikas.ai.framework.node.NodeStatus;
import com.ikas.ai.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Objects;

/**
 * @Description: 节点抽象
 * @Author: qi.yilai
 * @Date: 2023/1/12 17:08
 * @Version: 1.0.0
 **/
@Slf4j
public abstract class AbstractNode<O> implements Node<NodeInputVariables, O> {

    private volatile long startTime;

    private volatile long endTime;

    private String name;

    private Long nodeId;

    private Long nodeInstanceId;

    protected O result;

    private NodeStatus nodeStatus = NodeStatus.CREATED;

    /**
     * 执行异常
     */
    private String executeException;

    /**
     * 告警信息
     */
    protected String warningMessage;

    /**
     * 上个节点的结果作为入参
     */
    protected NodeInputVariables previous;

    /**
     * 所属流程
     */
    private ScoreFlow owner;

    /**
     * 事件监听器
     */
    private List<NodeListener> listeners = Lists.newArrayList();

    public AbstractNode(ScoreFlow scoreFlow, Long nodeId, Long nodeInstanceId, String name) {
        this.owner = scoreFlow;
        this.nodeId = nodeId;
        this.nodeInstanceId = nodeInstanceId;
        this.name = name;
        listeners.add(new NodeLogListener());
    }

    @Override
    public boolean execute() {
        this.startTime = System.currentTimeMillis();
        this.nodeStatus = NodeStatus.STARTING;
        fireEvent(this.nodeStatus, new JSONObject());
        try {
            inputGlobalVariables();
            this.result = doExecute();
            this.nodeStatus = NodeStatus.SUCCESS;
        } catch (Exception exception) {
            log.error("execute error", exception);
            executeException = exception.getMessage();
            this.nodeStatus = NodeStatus.FAILURE;
        } finally {
            this.endTime = System.currentTimeMillis();
            fireEvent(this.nodeStatus, new JSONObject());
        }
        return nodeStatus == NodeStatus.SUCCESS;
    }

    /**
     * 触发事件
     *
     * @param status
     * @param target
     */
    private void fireEvent(NodeStatus status, Object target) {
        NodeEvent nodeEvent = createNodeEvent(status, target);
        this.listeners.forEach(nodeListener -> {
            try {
                nodeListener.fireEvent(nodeEvent);
            } catch (Exception exception) {
                log.error("fire event error", exception);
            }
        });
    }

    private NodeEvent createNodeEvent(NodeStatus status, Object target) {
        NodeEvent nodeEvent = new NodeEvent();
        nodeEvent.setTarget(target);
        nodeEvent.setStatus(status);
        nodeEvent.setNow(System.currentTimeMillis());
        nodeEvent.setOwner(this);
        return nodeEvent;
    }

    @Override
    public void addListener(NodeListener nodeListener) {
        if (!listeners.contains(nodeListener)) {
            this.listeners.add(nodeListener);
        }
    }

    abstract O doExecute();

    @Override
    public NodeInputVariables previous() {
        return previous;
    }

    @Override
    public O result() {
        return result;
    }

    @Override
    public synchronized void input(NodeInputVariables i) {
        if (Objects.isNull(previous)) {
            this.previous = i;
        } else {
            previous.merge(i);
        }
    }

    public void inputGlobalVariables() {
        if (Objects.isNull(previous)) {
            this.previous = NodeInputVariables.empty();
        }
        this.previous.mergeParams(owner.getGlobalMap());
    }

    @Override
    public NodeInputVariables outVariables() {
        NodeInputVariables nodeInputVariables = NodeInputVariables.copy(previous);
        mergeData(nodeInputVariables, result);
        return nodeInputVariables;
    }

    protected void mergeData(NodeInputVariables nodeInputVariables, O result) {
        if (result instanceof Boolean) {
            nodeInputVariables.mergePredict((Boolean)result);
        } else if (result instanceof BatchDataFunction) {
            nodeInputVariables.mergeFunction((BatchDataFunction) result);
        } else if (result instanceof JSONObject) {
            nodeInputVariables.mergeParams((JSONObject) result);
        } else if (result instanceof HashMap) {
            nodeInputVariables.mergeParams((HashMap) result);
        }
    }

    @Override
    public Long nodeId() {
        return this.nodeId;
    }

    @Override
    public Long nodeInstanceId() {
        return this.nodeInstanceId;
    }

    @Override
    public long startTime() {
        return this.startTime;
    }

    @Override
    public long endTime() {
        return this.endTime;
    }

    @Override
    public String name() {
        return name;
    }

    @Override
    public NodeStatus status() {
        return nodeStatus;
    }

    @Override
    public ScoreFlow getOwner() {
        return owner;
    }

    @Override
    public String exception() {
        if (StringUtil.isEmpty(executeException)) {
            return null;
        } else {
            return executeException;
        }
    }

    @Override
    public String warningMessage() {
        if (StringUtil.isEmpty(warningMessage)) {
            return null;
        } else {
            return warningMessage;
        }
    }
}
